package com.xiguaji.tiktok.task.common.producer;

import com.gugee.tiktok.common.model.dal.blogger.BloggerLogRepository;
import com.gugee.tiktok.common.model.mysql.BloggerLog;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * Created with IntelliJ IDEA.
 * Description:查询需要的记录，构造消息并发送至主题中
 * User: luolifeng
 * Date: 2020-10-15 17:12
 */
@Slf4j
public abstract class AbstractProducerHandler<T> {
    @Autowired
    BloggerLogRepository bloggerLogRepository;


    public ReturnT<String> xxlJobStart(BloggerLog.BloggerLogType type, String who, String param, boolean topN){
        long start = System.currentTimeMillis();
        try {
            log.info("{}:开始启动消息投递任务: params={}",who, param);
            XxlJobLogger.log("{}:开始启动消息投递任务: params={}", who,param);
            TaskParam jobParam = JsonUtil.jsonParse(TaskParam.class, param);
            if (type==null || topN) {
                sendMsgWithAllTypes(jobParam);
            } else {
                int bloggerLogType = type.getValue();
                sendMsg2Kafka(bloggerLogType, who, jobParam);
            }

            return ReturnT.SUCCESS;
        } catch (Exception ex) {
            XxlJobLogger.log(ex);
            return ReturnT.FAIL;
        }finally {
            log.info("{}:投递结束  end cost total={},",who, (System.currentTimeMillis() - start) / 1000);
            XxlJobLogger.log("{}:投递结束  end cost total={}, params={}", who,(System.currentTimeMillis() - start) / 1000,param);
        }
    }
    /**
     * 获取当前需采集的页数
     * @param bloggerLogType 字典类型值，对应某种业务数据类别
     * @param who 操作方
     * @return
     */
   public BloggerLog getBloggerLog(int bloggerLogType,String who) {

        Optional<BloggerLog> bloggerLogOptional = bloggerLogRepository.findByLogType(bloggerLogType);
        BloggerLog bloggerLog;
        if (bloggerLogOptional.isPresent()) {
            bloggerLog = bloggerLogOptional.get();
        } else {
            bloggerLog = new BloggerLog();
            bloggerLog.setLogType(bloggerLogType);
            bloggerLog.setCreateBy(who);
            bloggerLog.setUpdateBy(who);
            bloggerLog.setCurrentBloggerId(1);
        }
        return bloggerLog;
    }

    /**
     * 查询具体页内的记录
     * @param bloggerLog 业务类型数据的采集页
     * @param taskParam  任务参数，包含了每页的页码大小
     * @return
     */
    public abstract List<T> findMessageObject(BloggerLog bloggerLog,TaskParam taskParam);
    /**
     * 查询Top排行N
     * @param taskParam  任务参数，包含了TopN大小
     * @return
     */
    public   List<T> findMessageObjectTopN(TaskParam taskParam){
        return new ArrayList<>();
    };
    public void sendMsgWithAllTypes(TaskParam taskParam){

    }
    public abstract void sendMsg(List<T> messages,String topic);

    /**
     * 发送
     * @param bloggerLogType
     * @param who
     * @param taskParam
     */
    public void sendMsg2Kafka(int bloggerLogType, String who,TaskParam taskParam){
        long start=System.currentTimeMillis();
        //获取将发送的位置
        BloggerLog bloggerLog=getBloggerLog(bloggerLogType,who);
        if(taskParam.getStartPage()==bloggerLog.getCurrentBloggerId()){
            String timeStr2= LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            bloggerLog.setFirstPageStartTime(timeStr2);
        }
        //获取应该发送的消息
        List<T> messages= findMessageObject(bloggerLog,taskParam);
        //记录下一次发送的位置
        long nextPage=bloggerLog.getCurrentBloggerId()%taskParam.getPagesLimit()+1;
        if (CollectionUtils.isEmpty(messages) || nextPage==taskParam.getStartPage()) {
            nextPage=taskParam.getStartPage();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            try {
                long mills=System.currentTimeMillis()-sdf.parse(bloggerLog.getFirstPageStartTime()).getTime();
                bloggerLog.setMoreInfo( "起始页="+taskParam.getStartPage()+",需轮询总页数="+bloggerLog.getCurrentBloggerId()+",上一轮花费(分钟)="+mills/1000/60+",上一轮开始时间="+bloggerLog.getFirstPageStartTime());
                log.warn("页码={} ，无数据,下一次将从{}页开始",bloggerLog.getCurrentBloggerId(),nextPage);
                XxlJobLogger.log("页码={} ，无数据,下一次将从{}页开始",bloggerLog.getCurrentBloggerId(),nextPage);
            } catch (ParseException e) {
                e.printStackTrace();
            }

        }


        sendMsg(messages,taskParam.getTopic());
        //发送消息后，设置下一次发送位置
        bloggerLog.setCurrentBloggerId(nextPage);
        bloggerLogRepository.save(bloggerLog);
        log.info(" who={}, size={},page={}, pageSize={},  record from  {} to {},times={}秒",who,messages.size(),bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),(bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),bloggerLog.getCurrentBloggerId()*taskParam.getRecords(),(System.currentTimeMillis()-start)/1000);
        XxlJobLogger.log("  size={},page={}, pageSize={},  record from  {} to {},times={}秒",messages.size(),bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),(bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),bloggerLog.getCurrentBloggerId()*taskParam.getRecords(),(System.currentTimeMillis()-start)/1000);

    }
//    public void sendMsg2KafkaTopN(String who,TaskParam taskParam){
//        long start=System.currentTimeMillis();
//        //获取应该发送的消息
//        List<T> messages= findMessageObjectTopN(taskParam);
//        sendMsg(messages,taskParam.getTopic());
//        log.info(" who={}, size={},topN={}, times={}秒",who,messages.size(), taskParam.getRecords(), (System.currentTimeMillis()-start)/1000);
//        XxlJobLogger.log("  size={},topN={},  times={}秒",messages.size(), taskParam.getRecords(), (System.currentTimeMillis()-start)/1000);
//
//    }
}
